package com.atguigu1.core.acc

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.util.AccumulatorV2
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable

/**
 *
 * @description: 广播变量可以将闭包的数据保存到Executor数据中,分布式共享只读变量.闭包数据都是以Task为单位发送的,每个人物中包含闭包数据
 *               这样会导致一个Executor中含有大量重复的数据,并且占用大量内存.Eexcutor其实就是一个JVM,所以在启动是会自动分配内存
 *               完全可以将任务助攻的闭包数据放在Executor的内存中,达到共享的目的
 * @time: 2021-03-12 11:45
 * @author: baojinlong
 **/
object Spark03BoardCast {

  // () => String 是一个函数，=> Unit 是一个执行结果为String的表达式
  def strToInt(s: => String): Int = {
    println("call strToInt")
    val result: Int = s.toInt
    println("result " + result)
    result
  }

  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("persistDemo")
    // 设置rdd分区数字
    val sparkContext = new SparkContext(conf)

    val rdd1: RDD[(String, Int)] = sparkContext.makeRDD(Seq(("a", 1), ("b", 2), ("c", 3)))
    val rdd2: RDD[(String, Int)] = sparkContext.makeRDD(Seq(("a", 4), ("b", 5), ("c", 6)))

    rdd1.join(rdd2).collect.foreach(println)
    println("**************")
    val commonMap: mutable.Map[String, Int] = mutable.Map(("a", 4), ("b", 5), ("c", 6))
    // 放置到广播变量中
    val bcMap: Broadcast[mutable.Map[String, Int]] = sparkContext.broadcast[mutable.Map[String, Int]](commonMap)

    rdd1.map {
      case (word, count) =>
        val wordCount: Int = commonMap.getOrElse(word, 0)
        // 访问广播变量
        val wordCount2: Int = bcMap.value.getOrElse(word, 0)
        // 作为结果返回
        (word, (count, wordCount))
    }.collect.foreach(println)


    strToInt(
      {
        println("eval parameter expression")
        "123"
      })

    sparkContext.stop
  }
}